package com.yb0os1.timewheel;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;

public class TimeWheelMqscQueue {
    private volatile long startTime;
    private MqscQueue[] wheel;
    private long cyberTime;//ms
    private final AtomicBoolean started;
    private Ticker ticker;

    private ThreadPoolExecutor executor;

    private final CountDownLatch startTimeLatch;

    public TimeWheelMqscQueue() {
        started = new AtomicBoolean(false);
        cyberTime = 100L;
        wheel = new MqscQueue[10];
        ticker = new Ticker();
        executor = new ThreadPoolExecutor(16, 32, 10, java.util.concurrent.TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());
        startTimeLatch = new CountDownLatch(1);
        for (int i = 0; i < wheel.length; i++) {
            wheel[i] = new MqscQueue();
        }
    }

    public void start() {
        if (started.compareAndSet(false, true)) {
            ticker.start();
        }
        try {
            startTimeLatch.await();//等待ticker线程为startTime赋值
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public void pushDelayTask(Runnable task, long delayMs) {
        //如果没有开启时间轮 需要开启
        start();
        //创建任务
        DelayTask delayTask = new DelayTask(task, delayMs);
        //计算任务应该所在下标
        int index = Math.toIntExact(((delayTask.deadline - startTime) / cyberTime) % wheel.length);
        //取出下标所在队列
        MqscQueue mqscQueue = wheel[index];
        mqscQueue.pushDelayTask(delayTask);
    }

    public void stop() {
        if (started.compareAndSet(true, false)) {
            LockSupport.unpark(ticker);
        }
    }


    private class MqscQueue {

        private final AtomicReference<DelayTask> head = new AtomicReference<>(null);

        public void pushDelayTask(DelayTask delayTask) {
            //头节点为空
            while (true) {
                DelayTask oldHead = head.get();
                delayTask.next = oldHead;
                if (head.compareAndSet(oldHead, delayTask)) {
                    return;
                }
            }
        }

        public List<Runnable> removeAndGetCanRunTask(long tickTime) {
            List<Runnable> tasks = new ArrayList<>();
            DelayTask current = head.get();
            DelayTask pre = null;
            while (current != null) {
                DelayTask next = current.next;
                if (current.deadline > tickTime) {//任务还不应该执行
                    pre = current;
                    current = next;
                    continue;
                }
                //任务应该执行了
                //非头节点
                if (pre != null) {
                    tasks.add(current.task);
                    pre.next = next;
                    current.next = null;
                    current = next;
                    continue;
                }
                //是头节点 交换头节点
                if (head.compareAndSet(current, next)) {
                    tasks.add(current.task);
                    current.next = null;
                    current = next;
                    continue;
                }
                //头节点交换失败 重试
                current = head.get();
            }
            return tasks;
        }

    }

    private class DelayTask {
        long deadline;
        Runnable task;

        DelayTask next;

        public DelayTask(Runnable task, long delayMs) {
            this.deadline = System.currentTimeMillis() + delayMs;
            this.task = task;
        }
    }

    private class Ticker extends Thread {
        private int tickCount = 0;

        @Override
        public void run() {
            startTime = System.currentTimeMillis();
            startTimeLatch.countDown();
            while (started.get()) {
                //获取当前真正执行任务的时间
                long tickTime = startTime + (tickCount + 1) * cyberTime;
                while (System.currentTimeMillis() <= tickTime) {
                    LockSupport.parkUntil(tickTime);
                    if (!started.get()) {
                        return;
                    }
                }
                //可以执行任务
                //获取下标
                int index = tickCount % wheel.length;
                MqscQueue queue = wheel[index];
                List<Runnable> tasks = queue.removeAndGetCanRunTask(tickTime);
                tasks.forEach(executor::execute);
                tickCount++;
            }
        }
    }
}
